package repo

import (
	"git.multiverse.io/eventkit/dxc-server/config"
	"git.multiverse.io/eventkit/dxc-server/constant"
	"git.multiverse.io/eventkit/dxc-server/models"
	"git.multiverse.io/framework/log"
	"github.com/go-errors/errors"
	"github.com/hazelcast/hazelcast-go-client"
	"strconv"
	"strings"
	"time"
)

// hazelcast data structure:
// use map, map key: `nodeId`:`rootXid`
// hash key: regTm                              	   hash value: RootTxn Register Timestamp
// hash key: txSts                                     hash value: RootTxn Status (0-trying, 1-confirming, 2-confirmed, 3-cancelling, 4-cancelled)
// hash key: stRcTm                      	   		   hash value: RootTxn Start Recovery Timestamp (only recovery use)
// hash key: @`branchXid`                              hash value: branch info (json, TsInfo struct)
// hash key: `branchXid`_sts                           hash value: branch status (0-trying, 1-confirmed, 2-cancelled)
// hash key: `branchXid`_rcTms                  	   hash value: branch recoverTimes
// hash key: `branchXid`_lstUpTm                	   hash value: branch LastUpdateTime (only recover use)
// hash key: `branchXid`_igCancel 				  	   hash value: branch TryFailedIgnoreCallbackCancel
// hash key: hasExpTm                         	  	   hash value: RootTxn HasExpireTime
type HazelcastRepository struct {
	client hazelcast.Client
	nodeId string
}

func NewHazelcastRepository() *HazelcastRepository {
	log.Infos("Start initialize hazelcast repository...")
	defer log.Infos("Successfully hazelcast repository!")

	dxcConfig := config.Cfg
	hzConfig := hazelcast.NewConfig()
	networkConfig := hzConfig.NetworkConfig()
	networkConfig.AddAddress(dxcConfig.Repo.Hazelcast.Addr...)
	client, err := hazelcast.NewClientWithConfig(hzConfig)
	if err != nil {
		panic(err)
	}
	return &HazelcastRepository{
		client: client,
		nodeId: dxcConfig.NodeId,
	}
}

func (h *HazelcastRepository) AddRoot(transaction *models.Transaction) (int, error) {
	tsInfo := &TsInfo{
		Xid:       transaction.BranchXid,
		PXid:      transaction.ParentXid,
		PrtAddr:   transaction.ParticipantAddress,
		SrvNm:     transaction.ServiceName,
		Param:     transaction.ParamData,
		StTm:      transaction.BranchTxnStartTime,
		CliSDKVer: transaction.ClientSDKVersion,
		Env:       transaction.Environment,
		Wks:       transaction.Workspace,
	}
	str, err := json.MarshalToString(tsInfo)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} marshal json fail, err:%s", transaction.RootXid, err)
		return 0, err
	}
	key := h.nodeId + ":" + transaction.BranchXid
	hRegisterTimeKey := constant.CACHE_REGISTER_TIME_KEY
	hRegisterTimeVal := h.timeToString(time.Now())
	hTxStatusKey := constant.CACHE_TX_STATUS_KEY
	hTxStatusVal := constant.ROOT_TXN_STATUS_TRYING
	hKey := constant.CACHE_BRANCH_KEY_PREFIX + transaction.BranchXid
	hVal := str
	hStatusKey := transaction.BranchXid + constant.CACHE_BRANCH_STATUS_SUFFIX_KEY
	hStatusVal := constant.BRANCH_TXN_STATUS_TRYING

	iMap, err := h.client.GetMap(key)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} save hazelcast fail, err:%s", transaction.BranchXid, err)
		return 0, err
	}
	mp := make(map[interface{}]interface{})
	mp[hRegisterTimeKey] = hRegisterTimeVal
	mp[hTxStatusKey] = hTxStatusVal
	mp[hKey] = hVal
	mp[hStatusKey] = hStatusVal
	err = iMap.PutAll(mp)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} save hezalcast fail, err:%s", transaction.BranchXid, err)
		return 0, err
	}
	return 1, nil
}

func (h *HazelcastRepository) AddBranch(transaction *models.Transaction) (int, error) {
	tsInfo := &TsInfo{
		Xid:       transaction.BranchXid,
		PXid:      transaction.ParentXid,
		PrtAddr:   transaction.ParticipantAddress,
		SrvNm:     transaction.ServiceName,
		Param:     transaction.ParamData,
		StTm:      transaction.BranchTxnStartTime,
		CliSDKVer: transaction.ClientSDKVersion,
		Env:       transaction.Environment,
		Wks:       transaction.Workspace,
	}
	str, err := json.MarshalToString(tsInfo)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} marshal json fail, err:%s", transaction.RootXid, err)
		return 0, err
	}
	hKey := constant.CACHE_BRANCH_KEY_PREFIX + transaction.BranchXid
	hVal := str
	key := h.nodeId + ":" + transaction.RootXid
	hStatusKey := transaction.BranchXid + constant.CACHE_BRANCH_STATUS_SUFFIX_KEY
	hStatusVal := constant.BRANCH_TXN_STATUS_TRYING

	iMap, err := h.client.GetMap(key)
	if err != nil {
		err = errors.Errorf("transaction{BranchXid:[%s]} save hazelcast fail, err:%s", transaction.BranchXid, err)
		return 0, err
	}
	mp := make(map[interface{}]interface{})
	mp[hKey] = hVal
	mp[hStatusKey] = hStatusVal
	err = iMap.PutAll(mp)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} save hezalcast fail, err:%s", transaction.RootXid, err)
		return 0, err
	}
	return 1, nil
}

func (h *HazelcastRepository) Recovery(root *models.RootTransaction) (int, error) {
	txns := make(map[interface{}]interface{}, 0)
	hRegisterTimeKey := constant.CACHE_REGISTER_TIME_KEY
	hRegisterTimeVal := h.timeToString(root.RootTxnRegisterTime)
	txns[hRegisterTimeKey] = hRegisterTimeVal
	hTxStatusKey := constant.CACHE_TX_STATUS_KEY
	hTxStatusVal := root.RootTxnStat
	txns[hTxStatusKey] = hTxStatusVal
	for _, branch := range root.Branches {
		tsInfo := &TsInfo{
			Xid:       branch.BranchXid,
			PXid:      branch.ParentXid,
			PrtAddr:   branch.ParticipantAddress,
			SrvNm:     branch.ServiceName,
			Param:     branch.ParamData,
			StTm:      branch.BranchTxnStartTime,
			CliSDKVer: branch.ClientSDKVersion,
			Wks:       branch.Workspace,
			Env:       branch.Workspace,
		}
		str, err := json.MarshalToString(tsInfo)
		if err != nil {
			err = errors.Errorf("transaction{RootXid:[%s]} marshal json fail, err:%s", branch.RootXid, err)
			return 0, err
		}
		hKey := constant.CACHE_BRANCH_KEY_PREFIX + branch.BranchXid
		hVal := str
		hStatusKey := branch.BranchXid + constant.CACHE_BRANCH_STATUS_SUFFIX_KEY
		hStatusVal := branch.BranchTxnStat
		txns[hKey] = hVal
		txns[hStatusKey] = hStatusVal
	}
	key := h.nodeId + ":" + root.RootXid

	iMap, err := h.client.GetMap(key)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} save hezalcast fail, err:%s", root.RootXid, err)
		return 0, err
	}
	err = iMap.PutAll(txns)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} save hezalcast fail, err:%s", root.RootXid, err)
		return 0, err
	}
	return 1, nil
}

func (h *HazelcastRepository) FindAllBranchByRootXid(rootXid string) (*models.RootTransaction, error) {
	key := h.nodeId + ":" + rootXid
	iMap, err := h.client.GetMap(key)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} hezalcast GetMap fail, err:%s", rootXid, err)
		return nil, err
	}
	entrySet, err := iMap.EntrySet()
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} hezalcast EntrySet fail, err:%s", rootXid, err)
		return nil, err
	}
	mp := make(map[string]string)
	for _, pair := range entrySet {
		key := pair.Key().(string)
		val := pair.Value().(string)
		mp[key] = val
	}
	if len(mp) == 0 {
		err = errors.Errorf("transaction{RootXid:[%s]} this transaction not exist", rootXid)
		return nil, err
	}
	hRegisterTimeKey := constant.CACHE_REGISTER_TIME_KEY
	registerTime, err := h.stringToTime(mp[hRegisterTimeKey])
	if err != nil {
		return nil, err
	}
	hTxStatusKey := constant.CACHE_TX_STATUS_KEY
	var startRecoverTime time.Time
	hStartRecoverTimeKey := constant.CACHE_START_RECOVER_TIME_KEY
	if startRecoverTimeStr, ok := mp[hStartRecoverTimeKey]; ok {
		startRecoverTime, err = h.stringToTime(startRecoverTimeStr)
		if err != nil {
			return nil, err
		}
	}
	var hasExpireTime bool
	hHasExpireTimeKey := constant.CACHE_HAS_EXPIRE_TIME_KEY
	if hHasExpireTimeVal, ok := mp[hHasExpireTimeKey]; ok {
		hasExpireTime, err = strconv.ParseBool(hHasExpireTimeVal)
		if err != nil {
			return nil, err
		}
	}
	rootTransaction := &models.RootTransaction{
		RootXid:                 rootXid,
		RootTxnStat:             mp[hTxStatusKey],
		RootTxnRegisterTime:     registerTime,
		RootTxnStartRecoverTime: startRecoverTime,
		HasExpireTime:           hasExpireTime,
	}
	// struct branches
	branchXIds := make([]string, 0)
	for k := range mp {
		if strings.HasPrefix(k, constant.CACHE_BRANCH_KEY_PREFIX) {
			branchXIds = append(branchXIds, k[constant.LENGTH_OF_CACHE_BRANCH_KEY_PREFIX:])
		}
	}
	transactions := make([]*models.Transaction, 0)
	for _, branchXId := range branchXIds {
		hBranchStatusKey := branchXId + constant.CACHE_BRANCH_STATUS_SUFFIX_KEY
		hRecoverTimesKey := branchXId + constant.CACHE_BRANCH_RECOVER_TIMES_SUFFIX_KEY
		recoveryTimes := 0
		if _, ok := mp[hBranchStatusKey]; ok {
			recoveryTimes, _ = strconv.Atoi(mp[hRecoverTimesKey])
		}
		tsInfo := new(TsInfo)
		tsInfoStr, ok := mp[constant.CACHE_BRANCH_KEY_PREFIX+branchXId]
		if !ok {
			return nil, errors.Errorf("transaction{RootXid:[%s], BranchXid:[%s]} this branch transaction not exist", rootXid, branchXId)
		}
		if err := json.UnmarshalFromString(tsInfoStr, tsInfo); err != nil {
			err = errors.Errorf("transaction{RootXid:[%s], BranchXid:[%s]} Unmarshal json fail, err:%s, mp[branchXId]=[%s]", rootXid, branchXId, err, mp[branchXId])
			return nil, err
		}
		transaction := &models.Transaction{
			RootXid:            rootXid,
			ParentXid:          tsInfo.PXid,
			BranchXid:          branchXId,
			ParticipantAddress: tsInfo.PrtAddr,
			BranchTxnStat:      mp[hBranchStatusKey],
			RecoveryTimes:      recoveryTimes,
			ServiceName:        tsInfo.SrvNm,
			ParamData:          tsInfo.Param,
			BranchTxnStartTime: tsInfo.StTm,
			ClientSDKVersion:   tsInfo.CliSDKVer,
		}
		hLastUpdateTimeKey := branchXId + constant.CACHE_BRANCH_LAST_UPDATE_TIME_SUFFIX_KEY

		if lastUpdateTimeStr, ok := mp[hLastUpdateTimeKey]; ok {
			lastUpdateTime, err := h.stringToTime(lastUpdateTimeStr)
			if err != nil {
				return nil, err
			}
			transaction.BranchTxnLastUpdateTime = lastUpdateTime
		}
		hTryFailedIgnoreCallbackCancelKey := branchXId + constant.CACHE_BRANCH_TRY_FAILED_IGNORE_CALLBACK_CANCEL_SUFFIX_KEY
		if tryFailedIgnoreCallbackCancelStr, ok := mp[hTryFailedIgnoreCallbackCancelKey]; ok {
			tryFailedIgnoreCallbackCancel, err := strconv.ParseBool(tryFailedIgnoreCallbackCancelStr)
			if err != nil {
				return nil, err
			}
			transaction.TryFailedIgnoreCallbackCancel = tryFailedIgnoreCallbackCancel
		}
		transactions = append(transactions, transaction)
	}
	rootTransaction.Branches = transactions
	return rootTransaction, nil
}

func (h *HazelcastRepository) DeleteTxn(rootXid string) (int, error) {
	key := h.nodeId + ":" + rootXid
	iMap, err := h.client.GetMap(key)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} hezalcast GetMap fail, err:%s", rootXid, err)
		return -1, err
	}
	err = iMap.Clear()
	return 0, nil
}

func (h *HazelcastRepository) IsExistsRootTransaction(rootXid string) (bool, error) {
	key := h.nodeId + ":" + rootXid
	iMap, err := h.client.GetMap(key)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} hezalcast GetMap fail, err:%s", rootXid, err)
		return false, err
	}
	hKey := constant.CACHE_BRANCH_KEY_PREFIX + rootXid
	exists, err := iMap.ContainsKey(hKey)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} exist root tx hezalcast HDel fail, err:%s", rootXid, err)
		return false, err
	}
	return exists, nil
}

func (h *HazelcastRepository) IsExistsBranchTransaction(rootXid string, branchXid string) (bool, error) {
	key := h.nodeId + ":" + rootXid
	iMap, err := h.client.GetMap(key)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} hezalcast GetMap fail, err:%s", rootXid, err)
		return false, err
	}
	hKey := constant.CACHE_BRANCH_KEY_PREFIX + branchXid
	exists, err := iMap.ContainsKey(hKey)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s],BranchXid: [%s]} HExist fail, err:%s", rootXid, branchXid, err)
		return false, err
	}
	return exists, nil
}

func (h *HazelcastRepository) TryCallback(rootXid string, callbackType string) (*models.RootTransaction, error) {
	// check exist
	exists, err := h.IsExistsRootTransaction(rootXid)
	if err != nil {
		return nil, err
	}
	if !exists {
		err = errors.Errorf("Register transaction{RootXid:[%s]} this root transaction not exist", rootXid)
		return nil, err
	}
	key := h.nodeId + ":" + rootXid
	hTxStatusKey := constant.CACHE_TX_STATUS_KEY
	iMap, err := h.client.GetMap(key)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} hezalcast GetMap fail, err:%s", rootXid, err)
		return nil, err
	}
	_, err = iMap.Put(hTxStatusKey, callbackType)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s], callbackType:[%s]} set root tx status hezalcast HSet fail, err:%s", rootXid, callbackType, err)
		return nil, err
	}

	result, err := h.FindAllBranchByRootXid(rootXid)
	if err != nil {
		return nil, err
	}
	return result, nil
}

func (h *HazelcastRepository) TryEnList(rootXid string) error {
	key := h.nodeId + ":" + rootXid
	hTxStatusKey := constant.CACHE_TX_STATUS_KEY
	iMap, err := h.client.GetMap(key)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} hezalcast GetMap fail, err:%s", rootXid, err)
		return err
	}
	value, err := iMap.Get(hTxStatusKey)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} get root tx status hezalcast HGet fail, err:%s", rootXid, err)
		return err
	}
	if value.(string) != "0" {
		return errors.Errorf("TryEnList failed, Invalid root transaction stat[%s]", rootXid)
	}
	return nil
}

func (h *HazelcastRepository) CallbackFail(rootXid string, branches []*models.Transaction, firstTimeCallback bool) (int, error) {
	key := h.nodeId + ":" + rootXid
	mp := make(map[interface{}]interface{}, 0)
	if firstTimeCallback {
		hStartRecoverTimeKey := constant.CACHE_START_RECOVER_TIME_KEY
		hStartRecoverTimeVal := h.timeToString(time.Now())
		mp[hStartRecoverTimeKey] = hStartRecoverTimeVal
	}
	for _, branch := range branches {
		hStatusKey := branch.BranchXid + constant.CACHE_BRANCH_STATUS_SUFFIX_KEY
		hStatusVal := branch.BranchTxnStat
		hRecoverTimesKey := branch.BranchXid + constant.CACHE_BRANCH_RECOVER_TIMES_SUFFIX_KEY
		hRecoverTimesVal := strconv.Itoa(branch.RecoveryTimes)
		hLastUpdateTimeKey := branch.BranchXid + constant.CACHE_BRANCH_LAST_UPDATE_TIME_SUFFIX_KEY
		hLastUpdateTimeVal := strconv.FormatInt(branch.BranchTxnLastUpdateTime.Unix(), 10)
		mp[hStatusKey] = hStatusVal
		mp[hRecoverTimesKey] = hRecoverTimesVal
		mp[hLastUpdateTimeKey] = hLastUpdateTimeVal
	}

	iMap, err := h.client.GetMap(key)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} hezalcast GetMap fail, err:%s", rootXid, err)
		return 0, err
	}
	err = iMap.PutAll(mp)
	if err != nil {
		return 0, err
	}
	return len(branches), nil
}

func (h *HazelcastRepository) FindRootBranchesByTimeExpire(expireSeconds int64) (rootXIds []*models.Transaction, err error) {
	panic("implement me")
}

func (h *HazelcastRepository) UpdateBranchTryFailedIgnoreCallbackCancel(rootXid string, branchXid string) (int, error) {
	key := h.nodeId + ":" + rootXid
	hTryFailedIgnoreCallbackCancelKey := branchXid + constant.CACHE_BRANCH_TRY_FAILED_IGNORE_CALLBACK_CANCEL_SUFFIX_KEY
	hTryFailedIgnoreCallbackCancelVal := strconv.FormatBool(true)

	iMap, err := h.client.GetMap(key)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} hezalcast GetMap fail, err:%s", rootXid, err)
		return 0, err
	}
	_, err = iMap.Put(hTryFailedIgnoreCallbackCancelKey, hTryFailedIgnoreCallbackCancelVal)
	if err != nil {
		return 0, err
	}
	return 1, nil
}

func (h *HazelcastRepository) ExpireTime(rootXid string) (int, error) {
	key := h.nodeId + ":" + rootXid
	hHasExpireTimeKey := constant.CACHE_HAS_EXPIRE_TIME_KEY
	hHasExpireTimeVal := strconv.FormatBool(true)

	iMap, err := h.client.GetMap(key)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} hezalcast GetMap fail, err:%s", rootXid, err)
		return 0, err
	}
	_, err = iMap.Put(hHasExpireTimeKey, hHasExpireTimeVal)
	if err != nil {
		return 0, err
	}
	return 1, nil
}

func (h *HazelcastRepository) LoadRootsFromPersistenceRepo(count int64) ([]string, error) {
	return nil, nil
}

func (h *HazelcastRepository) timeToString(t time.Time) string {
	return strconv.FormatInt(t.Unix(), 10)
}

func (h *HazelcastRepository) stringToTime(s string) (time.Time, error) {
	timestamp, err := strconv.ParseInt(s, 10, 64)
	if err != nil {
		err = errors.Errorf("string to time fail, err:%s", err)
	}
	return time.Unix(timestamp, 0), nil
}
